#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import datetime
import grp
import json
import os
import pwd
import re
import socket
import stat
import time
import uuid
import subprocess
import shlex

import pexpect
import psutil

from common import common
from common.common import output_execution_result_ex, execute_cmd, output_result_file, \
    execute_cmd_oversize_return_value, clean_dir, check_command_injection, check_path_legal, \
    check_command_injection_exclude_quote
from common.common_models import ActionResult, SubJobDetails, LogDetail
from common.const import CMDResult, ParamConstant, DBLogLevel, SysData, SubJobStatusEnum, Encoding, AuthType
from common.logger import Logger
from common.number_const import NumberConst
from db2.backup.util.ha_utils import is_power_rhel_ha
from db2.comm.const import Db2Const, RpcToolApiName
from db2.comm.db2_exception import ErrCodeException
from db2.comm.error_code import Db2ErrCode
from db2.comm.constant import ParamField
from db2.comm.models.backup_models import LogBackupParam
from db2.comm.util.comm_decorators import handle_ex_return_default_value

LOGGER = Logger().get_logger(filename="db2.log")


class Db2CommonUtil:
    """
    DB2通用工具类
    """
    @staticmethod
    def check_path_characters_and_black_list(path):
        if check_command_injection(path):
            LOGGER.error(f"String contains special characters!")
            return False
        if re.search(Db2Const.PATH_BLACK_LIST, path):
            LOGGER.error(f"Path in black list!")
            return False
        return True

    @staticmethod
    def get_group_name_by_os_user(os_user):
        """根据用户名获取用户组名"""
        group_id = pwd.getpwnam(os_user).pw_gid
        return grp.getgrgid(group_id).gr_name

    @staticmethod
    def get_group_id_by_os_user(os_user):
        """根据用户名获取用户组id"""
        return pwd.getpwnam(os_user).pw_gid

    @staticmethod
    def get_os_user_id_by_os_user(os_user):
        """根据用户名获取用户id"""
        return pwd.getpwnam(os_user).pw_uid

    @staticmethod
    def get_file_user_id_and_group_id(file_path):
        stat_info = os.stat(file_path)
        return stat_info.st_uid, stat_info.st_gid

    @staticmethod
    def get_local_ips() -> list:
        """
        获取本机所有IP
        """
        LOGGER.info(f"Start getting all local ips ...")
        local_ips = []
        ip_dict = psutil.net_if_addrs()
        for _, tmp_address_list in ip_dict.items():
            for i in tmp_address_list:
                if i.family == NumberConst.TWO and i.address != '127.0.0.1':
                    local_ips.append(i.address)
        LOGGER.info(f"Get all local ips: {local_ips} success.")
        return local_ips

    @staticmethod
    def create_soft_link(src_path, tgt_path):
        """创建软连接"""
        Db2CommonUtil.check_path_list(src_path, tgt_path)
        LOGGER.info(f"Starting create soft link, source path: {src_path}, target path: {tgt_path}.")
        try:
            os.symlink(src_path, tgt_path)
        except Exception as ex:
            LOGGER.exception("Create soft link failed")
            raise Exception("Create soft link failed") from ex
        LOGGER.info("Create soft link success")

    @staticmethod
    def get_db_cfg_for_db(os_user, db_name):
        """获取数据库的配置信息"""
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(db_name)
        get_db_cfg_cmd = f'su - {os_user} -c "db2 get db cfg for {db_name}"'
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, out, err = common.execute_cmd(get_db_cfg_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute get db cfg for {db_name} command failed, os user: {os_user}, "
                         f"return code: {return_code}, out: {out}, err: {err}.")
            raise Exception(f"Execute get db cfg for {db_name} command failed")
        LOGGER.info(f"Execute get db cfg for {db_name} command success.")
        return out

    @staticmethod
    def get_db_cfg_by_pattern(os_user, db_name, pattern: list):
        """查询数据库配置"""
        if not pattern:
            LOGGER.error(f"Pattern is empty.")
            raise Exception("Pattern is empty")
        value = ""
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(db_name)
        encoding = Db2CommonUtil.get_lang_value(os_user)
        get_db_cfg_cmd = [f"su - {os_user} -c 'db2 get db cfg for {db_name}'"]
        return_code, out, err = Db2CommonUtil.execute_cmd_list(get_db_cfg_cmd, encoding=encoding)
        config_list = out.split("\n")

        if any([return_code != CMDResult.SUCCESS.value, not config_list, len(config_list) < 2]):
            LOGGER.error(f"Database: {db_name} does not have a configuration that matches pattern: {pattern[0]}, "
                         f"err info: {err}.")
            raise Exception("Database does not have a specified configuration")
        if "Database Configuration" in config_list[1] or len(pattern) == 1:
            for item in config_list:
                if pattern[0] in item:
                    value = str(item.split("=")[1]).strip()
                    break
        elif "数据库配置" in config_list[1] and len(pattern) == 2:
            for item in config_list:
                if pattern[1] in item:
                    value = str(item.split("=")[1]).strip()
                    break
        else:
            LOGGER.error(f"Get database: {db_name} configuration pattern: {pattern[0]} failed, out info: {out}, "
                         f"err info: {err}.")
            raise Exception("Database does not have a specified configuration")
        LOGGER.info(f"Database: {db_name} has a configuration that matches pattern: {pattern[0]}, value: {value}, "
                    f"os user: {os_user}.")
        return value

    @staticmethod
    def get_log_arch_meth_val_of_db(os_user, db_name):
        """查询数据库配置项LOGARCHMETH1的值"""
        log_arch_meth_val = Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["(LOGARCHMETH1)"])
        if ":" in log_arch_meth_val:
            return str(log_arch_meth_val.split(":")[1]).strip()
        return log_arch_meth_val

    @staticmethod
    def check_log_arch_meth_is_dir(log_arch_meth_val):
        if log_arch_meth_val and os.path.isdir(log_arch_meth_val):
            LOGGER.debug(f"The LOGARCHMETH cfg value: {log_arch_meth_val} of database is directory.")
            return True
        LOGGER.error(f"The LOGARCHMETH cfg value: {log_arch_meth_val} of database is not directory.")
        return False

    @staticmethod
    def get_track_mod_val_of_db(os_user, db_name):
        """查询数据库配置项TRACKMOD的值"""
        return Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["(TRACKMOD)"])

    @staticmethod
    def get_rollforward_pending_val_of_db(os_user, db_name):
        """查询数据库配置项Rollforward pending的值"""
        return Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["Rollforward pending", "前滚暂挂"])

    @staticmethod
    def get_restore_pending_val_of_db(os_user, db_name):
        """查询数据库配置项Restore pending的值"""
        return Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["Restore pending", "复原暂挂"])

    @staticmethod
    def get_log_path_val_of_db(os_user, db_name):
        """查询数据库配置项Path to log files的值"""
        return Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["Path to log files", "日志文件路径"])

    @staticmethod
    def get_first_active_log_file_of_db(os_user, db_name):
        """查询数据库配置项"First active log file"的值"""
        return Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["First active log file", "首个活动日志文件"])

    @staticmethod
    @handle_ex_return_default_value(default_return_value=Db2Const.LOG_ARCHIVE_MIN_TIMEOUT)
    def get_archive_timeout_of_db(os_user, db_name):
        """查询归档日志超时时间"""
        retry_num = Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["(NUMARCHRETRY)"])
        delay = Db2CommonUtil.get_db_cfg_by_pattern(os_user, db_name, ["(ARCHRETRYDELAY)"])
        LOGGER.info(f"Number of log archive retries on error is: {retry_num}, log archive retry delay is: {delay}.")
        try:
            return int(retry_num) * int(delay) + Db2Const.LOG_ARCHIVE_MIN_TIMEOUT
        except ValueError:
            LOGGER.error(f"Retry_num: {retry_num} or delay: {delay} is not number.")
            raise Exception("Retry_num or delay is not number.")

    @staticmethod
    def parse_all_database_info_to_dict(list_ts_ret: str):
        db_dict = dict()
        tmp_db_name = None
        for _, line in enumerate(list_ts_ret.split("\n")):
            if re.match(r"\s*Database name\s*=", str(line)):
                tmp_db_name = str(line).split("=")[1].strip()
                db_dict[tmp_db_name] = {}
                continue
            elif re.match(r"\s*数据库名称\s*=", str(line)):
                tmp_db_name = str(line).split("=")[1].strip()
                db_dict[tmp_db_name] = {}
                continue
            if not tmp_db_name:
                continue
            if re.match(r"\s*Local database directory\s*=", str(line)):
                db_dict[tmp_db_name]["localDatabaseDirectory"] = str(line).split("=")[1].strip()
                continue
            elif re.match(r"\s*本地数据库目录\s*=", str(line)):
                db_dict[tmp_db_name]["localDatabaseDirectory"] = str(line).split("=")[1].strip()
                continue
            if re.match(r"\s*Catalog database partition number\s*=", str(line)):
                db_dict[tmp_db_name]["catalogPartitionNumber"] = str(line).split("=")[1].strip()
            elif re.match(r"\s*目录数据库分区号\s*=", str(line)):
                db_dict[tmp_db_name]["catalogPartitionNumber"] = str(line).split("=")[1].strip()
        return db_dict

    @staticmethod
    def get_db_info_dict(os_user) -> dict:
        """获取所有的数据库信息字典"""
        Db2CommonUtil.check_os_user_with_ex(os_user)
        list_db_cmd = f'su - {os_user} -c "db2 list db directory"'
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, std_out, std_err = execute_cmd(list_db_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute list db directory command failed.")
            raise Exception("Execute list db command failed")
        if "System Database Directory" not in std_out and "系统数据库目录" not in std_out:
            LOGGER.error("The database directory information is incorrect.")
            raise Exception("The database directory information is incorrect")
        db_dict = Db2CommonUtil.parse_all_database_info_to_dict(std_out)
        LOGGER.info(f"Get database info success, result: {db_dict}.")
        return db_dict

    @staticmethod
    def filter_ts_info_dict(ts_dict):
        """
        过滤掉不支持备份的表空间信息：SYSCATSPACE、系统工具表空间SYSTOOLSPACE、系统临时表空间（如TEMPSPACE1）、用户临时表空间
        """
        filtered_ts_dict = dict()
        for k, tmp_ts_info in ts_dict.items():
            if k in ("SYSCATSPACE", "SYSTOOLSPACE"):
                continue
            if tmp_ts_info.get("contents") in ("System Temporary data", "User Temporary data"):
                continue
            filtered_ts_dict[k] = tmp_ts_info
        return filtered_ts_dict

    @staticmethod
    def get_table_space_info_dict_of_db(os_user, db_name) -> dict:
        """获取数据库的所有支持备份的表空间信息"""
        LOGGER.info(f"Getting tablespaces of database: {db_name}, os user: {os_user}.")
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(db_name)
        list_ts_cmd = f'su - {os_user} -c "db2 connect to {db_name}; db2 list tablespaces; db2 terminate"'
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, std_out, std_err = Db2CommonUtil.execute_cmd_oversize_return_value(list_ts_cmd, encoding)
        LOGGER.info(f"return_code:{return_code}, std_out:{std_out}, std_err:{std_err}")
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute list table spaces command failed, code: {return_code}, out: {std_out}, "
                         f"err: {std_err}.")
            raise Exception("Execute list table spaces command failed")

        if "Tablespaces for Current Database" in std_out:
            ts_dict = Db2CommonUtil.parse_all_tablespace_info_to_dict(std_out)
            filtered_ts_dict = Db2CommonUtil.filter_ts_info_dict(ts_dict)
        elif "当前数据库的表空间" in std_out:
            ts_dict = Db2CommonUtil.parse_all_chinese_tablespace_info_to_dict(std_out)
            filtered_ts_dict = Db2CommonUtil.filter_ts_info_dict(ts_dict)
        else:
            LOGGER.error(f"The listed database tablespace information is incorrect, out: {std_out}.")
            raise Exception("The listed database tablespace information is incorrect")
        LOGGER.info(f"Get table spaces of database: {db_name} success, all tablespace info: {ts_dict}, "
                    f"filtered tablespace info: {filtered_ts_dict}.")
        return filtered_ts_dict

    @staticmethod
    def parse_all_tablespace_info_to_dict(list_ts_ret: str):
        ts_dict = dict()
        tmp_ts_name = None
        for _, line in enumerate(list_ts_ret.split("\n")):
            if re.match(r"\s*Name\s*=", str(line)):
                tmp_ts_name = str(line).split("=")[1].strip()
                ts_dict[tmp_ts_name] = {}
                continue
            if not tmp_ts_name:
                continue
            if re.match(r"\s*Type\s*=", str(line)):
                ts_dict[tmp_ts_name]["type"] = str(line).split("=")[1].strip()
                continue
            if re.match(r"\s*Contents\s*=", str(line)):
                ts_dict[tmp_ts_name]["contents"] = str(line).split("=")[1].strip()
                continue
            if re.match(r"\s*State\s*=", str(line)):
                ts_dict[tmp_ts_name]["state"] = str(line).split("=")[1].strip()
        return ts_dict

    @staticmethod
    def parse_all_chinese_tablespace_info_to_dict(list_ts_ret: str):
        ts_dict = dict()
        tmp_ts_name = None
        for _, line in enumerate(list_ts_ret.split("\n")):
            if re.match(r"\s*名称\s*=", str(line)):
                tmp_ts_name = str(line).split("=")[1].strip()
                ts_dict[tmp_ts_name] = {}
                continue
            if not tmp_ts_name:
                continue
            if re.match(r"\s*类型\s*=", str(line)):
                ts_dict[tmp_ts_name]["type"] = str(line).split("=")[1].strip()
                continue
            if re.match(r"\s*内容\s*=", str(line)):
                ts_dict[tmp_ts_name]["contents"] = str(line).split("=")[1].strip()
                continue
            if re.match(r"\s*状态\s*=", str(line)):
                ts_dict[tmp_ts_name]["state"] = str(line).split("=")[1].strip()
        return ts_dict


    @staticmethod
    def list_applications_for_db(os_user, db_name):
        """获取连接指定数据库的Handle ID列表"""
        app_handles = list()
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(db_name)
        list_db_app_cmd = f"su - {os_user} -c 'db2 list applications for db {db_name}'"
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, out, err = execute_cmd(list_db_app_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.warning(f"Execute list applications for database: {db_name} command failed, "
                           f"return code: {return_code}, out: {out}, err: {err}.")
            return app_handles
        LOGGER.info(f"List applications for database {db_name} success.")
        # 样例：DB2INST1 db2bp          3045       *N0.db2inst1.230101120001                  SAMPLE      1
        match_conn_info_ret = r"\s*\S+\s+\S+\s+\S+\s+\S+\s+" + db_name + "\s+\S+"
        for line in out.split('\n'):
            if re.match(match_conn_info_ret, str(line), flags=re.I):
                tmp_handle = str(line).split()[2]
                app_handles.append(tmp_handle)
        LOGGER.info(f"Find the application connected to the database: {db_name} success, handle ids: {app_handles}.")
        return app_handles

    @staticmethod
    def force_application_by_handle_ids(os_user, handle_ids):
        if not handle_ids:
            return
        handle_str = ','.join(handle_ids)
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(handle_str)
        disconnect_db_cmd = f'su - {os_user} -c "db2 force application \({handle_str}\)"'
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, out, err = execute_cmd(disconnect_db_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute force application command failed, handle ids: {handle_ids}, "
                         f"return code: {return_code}, out: {out}, err: {err}.")
            return
        LOGGER.info(f"Execute force application command success, handle ids: {handle_ids}.")

    @staticmethod
    def disconnect_db_application(os_user, db_name, force_times=NumberConst.THREE):
        """端口数据库所有连接"""
        for _ in range(force_times):
            app_handles = Db2CommonUtil.list_applications_for_db(os_user, db_name)
            if not app_handles:
                break
            Db2CommonUtil.force_application_by_handle_ids(os_user, app_handles)
            time.sleep(NumberConst.TEN)
        LOGGER.info(f"Disconnect application of database success, database name: {db_name}.")

    @staticmethod
    def kill_agent_job_process(job_pid, job_id):
        """杀掉agent任务进程"""
        os_pid_list = psutil.pids()
        for tmp_os_pid in os_pid_list:
            tmp_process = psutil.Process(tmp_os_pid)
            tmp_cmd = tmp_process.cmdline()
            if "python3" in tmp_cmd and job_id in tmp_cmd and job_pid not in tmp_cmd:
                tmp_process.kill()
                LOGGER.info(f"The agent job has been terminated, os process id: {tmp_os_pid}, cmd: {tmp_cmd}"
                            f"job pid: {job_pid}, job id: {job_id}.")
                break

    @staticmethod
    def build_action_result(code, body_err=None, message=None, err_params: list = None):
        return ActionResult(code=code, bodyErr=body_err, message=message, bodyErrParams=err_params)

    @staticmethod
    def build_log_detail(log_label, log_info_params, err_code=None,
                         log_detail_param=None, log_level=DBLogLevel.INFO.value):
        log_info_params = log_info_params if log_info_params else list()
        return LogDetail(logInfo=log_label,
                         logInfoParam=log_info_params,
                         logTimestamp=int(time.time()),
                         logDetail=err_code,
                         logDetailParam=log_detail_param,
                         logLevel=log_level)

    @staticmethod
    def write_progress_to_file(s_job_detail: SubJobDetails, progress_f_name, cache_path, log_detail: LogDetail = None):
        """
        将进度信息写入进度文件供框架读取
        """
        if log_detail:
            s_job_detail.log_detail.append(log_detail)
        file_path = os.path.realpath(os.path.join(cache_path, progress_f_name))
        LOGGER.info(f"Write progress info, progress file: {file_path}, progress: {s_job_detail.progress}, "
                    f"task status: {s_job_detail.task_status}")
        output_execution_result_ex(file_path, s_job_detail.dict(by_alias=True))

    @staticmethod
    def output_action_result(pid, code, body_err=None, message=None, err_params: list = None):
        """
        将actionResult写入到结果文件,供框架读取
        :return:
        """
        json_result = ActionResult(code=code, bodyErr=body_err, message=message, bodyErrParams=err_params)
        LOGGER.info(f"Output action result: {json_result.dict(by_alias=True)}.")
        file_path = os.path.join(ParamConstant.RESULT_PATH, f"result{pid}")
        output_execution_result_ex(file_path, json_result.dict(by_alias=True))

    @staticmethod
    def remove_file_if_exists(file_path):
        LOGGER.info(f"Start removing file: {file_path} ...")
        if file_path and os.path.isfile(file_path) and Db2CommonUtil.check_path_characters_and_black_list(file_path):
            os.remove(file_path)
            LOGGER.info(f"Remove file: {file_path} success.")

    @staticmethod
    def clear_dir_if_exists(dir_path):
        LOGGER.info(f"Start clearing the directory: {dir_path} ...")
        if not dir_path or not os.path.isdir(dir_path):
            LOGGER.warning(f"The path: {dir_path} is not directory when clearing it.")
            return
        clean_dir(dir_path)
        LOGGER.info(f"Clear the directory: {dir_path} success.")

    @staticmethod
    def clear_soft_link_of_dir(input_dir):
        """清理输入目录下的软连接"""
        LOGGER.info("Start clearing soft link of input directory.")
        if not os.path.exists(input_dir) or not os.path.isdir(input_dir):
            LOGGER.warning("The input directory does not exists or is not directory when clearing soft link.")
            return
        Db2CommonUtil.check_path_list(input_dir)
        for tmp_path in os.listdir(input_dir):
            tmp_abs_path = os.path.realpath(os.path.join(input_dir, tmp_path))
            if os.path.islink(tmp_abs_path):
                LOGGER.debug(f"Removing soft link: {tmp_abs_path}.")
                Db2CommonUtil.check_path_list(tmp_abs_path)
                os.remove(tmp_abs_path)
        LOGGER.info("Clear soft link of input directory success.")

    @staticmethod
    def get_db2_instance_name_by_os_name(os_name: str):
        # 一个用户只有一个实例，但是实例名可以和用户名不一致
        Db2CommonUtil.check_os_user_with_ex(os_name)
        var_cmd = f"su - {os_name} -c 'db2 get instance'"
        encoding = Db2CommonUtil.get_lang_value(os_name)
        return_code, out, err = execute_cmd(var_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"execute db2 get instance fail return code: {return_code}, out: {out}, err: {err}.")
            return ""
        LOGGER.info("Begin to parse instance")
        db2_instance = ""
        instance = out.split("\n")
        for instancename in instance:
            if instancename == "":
                continue
            split_instance_name = instancename.split(":")
            if len(split_instance_name) > 1:
                db2_instance = split_instance_name[1].strip()
        return db2_instance

    @staticmethod
    def clear_archive_log_dir(os_user, db_name, instance_name):
        log_arch_meth_val = Db2CommonUtil.get_log_arch_meth_val_of_db(os_user, db_name)
        if Db2CommonUtil.check_log_arch_meth_is_dir(log_arch_meth_val):
            arch_log_path = os.path.realpath(os.path.join(log_arch_meth_val, instance_name, db_name))
            Db2CommonUtil.remove_file_if_exists(arch_log_path)
            Db2CommonUtil.clear_dir_if_exists(arch_log_path)

    @staticmethod
    def exec_rpc_tool_cmd(api_name, in_path, out_path):
        rpc_tool_path = os.path.realpath(os.path.join(ParamConstant.BIN_PATH, 'rpctool.sh'))
        rpc_tool_cmd = f"{rpc_tool_path} {api_name} {in_path} {out_path}"
        LOGGER.info(f"Start executing rpc tool command: {rpc_tool_cmd}.")
        return_code, out, err = execute_cmd(rpc_tool_cmd)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute rpc tool command failed, return code: {return_code}, "
                         f"out: {out}, err: {err}.")
            return False
        LOGGER.info(f"Execute rpc tool command success.")
        return True

    @staticmethod
    def exec_rpc_tool_cmd_with_output(api_name, in_path, out_path):
        rpc_tool_path = os.path.realpath(os.path.join(ParamConstant.BIN_PATH, 'rpctool.sh'))
        rpc_tool_cmd = f"{rpc_tool_path} {api_name} {in_path} {out_path}"
        LOGGER.info(f"Start executing rpc tool command: {rpc_tool_cmd} with output ...")
        return_code, out, err = execute_cmd(rpc_tool_cmd)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute rpc tool command with output failed, return code: {return_code}, "
                         f"out: {out}, err: {err}.")
            return False, {}
        output_dict = Db2CommonUtil.read_json_from_file(out_path)
        LOGGER.info(f"Execute rpc tool command with output success.")
        return True, output_dict

    @staticmethod
    def convert_backup_time_to_timestamp(time_str):
        struct_time = time.strptime(time_str, "%Y%m%d%H%M%S")
        return int(time.mktime(struct_time))

    @staticmethod
    def convert_backup_time_to_timestamp_v2(time_str):
        try:
            struct_time = time.strptime(time_str[:-7], "%Y-%m-%d-%H.%M.%S")
        except Exception as e_info:
            return int(time_str)
        return int(time.mktime(struct_time))

    @staticmethod
    def convert_copy_timestamp_to_restore_time(s_timestamp):
        """将副本时间戳转换为恢复使用的时间"""
        try:
            s_timestamp = int(s_timestamp)
        except ValueError as ex:
            LOGGER.exception(f"Timestamp parameter: {s_timestamp} is invalid when converting.")
            raise Exception("Timestamp parameter is invalid") from ex
        return datetime.datetime.fromtimestamp(s_timestamp).strftime("%Y%m%d%H%M%S")

    @staticmethod
    def convert_pitr_timestamp_to_local_time(s_timestamp):
        """将PITR时间戳转换为rollforward使用的local time"""
        try:
            s_timestamp = int(s_timestamp)
        except ValueError as ex:
            LOGGER.exception(f"Timestamp parameter: {s_timestamp} is invalid when converting.")
            raise Exception("Timestamp parameter is invalid") from ex
        return datetime.datetime.fromtimestamp(s_timestamp).strftime("%Y-%m-%d-%H.%M.%S")

    @staticmethod
    def get_version(user_name):
        """获取DB2数据库版本号"""
        version = ""
        Db2CommonUtil.check_os_user_with_ex(user_name)
        get_version_cmd = f"su - {user_name} -c 'db2level'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        return_code, std_out, std_err = execute_cmd(get_version_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute db2level command failed when getting db2 version.")
            return version
        for i in std_out.split('\n'):
            ret = re.search(r"\"DB2\s+v(.*?)\"", i)
            if ret:
                version = ret.groups()[0]
                return str(version).strip()
        LOGGER.error(f"The obtained db2 version is empty.")
        return version

    @staticmethod
    def get_db_data_backup_end_time_and_cur_log(user_name, db_name, start_time, ptn_num=None, deploy_type=None):
        """
        获取数据库数据备份的结束时间，当前日志文件
        :param user_name: 用户名
        :param db_name: 数据库
        :param start_time: 数据备份开始时间
        :return: end_time, early_log, cur_log
        """
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(db_name, str(start_time), str(ptn_num))
        lis_bak_cmd = f"su - {user_name} -c 'db2 list history backup since {start_time} for db {db_name}'"
        if deploy_type == Db2Const.DPF_CLUSTER_TYPE and ptn_num is not None:
            lis_bak_cmd = f"su - {user_name} -c 'db2_all \"<<+{ptn_num}<db2 list history backup since {start_time} " \
                          f"for db {db_name}\"'"
        return_code, std_out, std_err = execute_cmd(lis_bak_cmd, encoding=Db2CommonUtil.get_lang_value(user_name))
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f'Execute list backup history command: {lis_bak_cmd} failed, return code: {return_code}, '
                         f'out: {std_out}, err: {std_err}')
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[lis_bak_cmd, std_out])
        find_flags = False
        # current log信息："  B  D  20230224093825001   N    D  S0000000.LOG S0000000.LOG"
        cur_log_begin_reg, cur_log_end_reg, cur_log_info_reg = Db2CommonUtil.assemble_regularized_expression(start_time)
        out_lines = std_out.split('\n')
        end_time, early_log, cur_log = "", "", ""
        for i, item in enumerate(out_lines):
            if all([end_time, early_log, cur_log]):
                break
            mat_cur_log_ret = re.search(cur_log_info_reg, item)
            if mat_cur_log_ret:
                early_log = mat_cur_log_ret.groups()[0]
                cur_log = mat_cur_log_ret.groups()[1]
                find_flags = True
                continue
            if find_flags and ("Start Time" in item or "开始时间" in item):
                try:
                    find_start_time = item.split(':')[1].strip()
                except Exception:
                    find_start_time = item.split('：')[1].strip()
                if start_time not in find_start_time:
                    err_msg = "Get data backup start time from backup history failed"
                    LOGGER.error(err_msg)
                    raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[lis_bak_cmd, err_msg])
                end_time_line = out_lines[i + 1]
                if "End Time" not in end_time_line and "结束时间" not in end_time_line:
                    err_msg = "Get data backup end time from backup history failed"
                    LOGGER.error(err_msg)
                    raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[lis_bak_cmd, err_msg])
                try:
                    end_time = end_time_line.split(':')[1].strip()
                except Exception:
                    end_time = end_time_line.split('：')[1].strip()
        if any([not end_time, not early_log, not cur_log]):
            err_msg = f"Get data backup end time: {end_time}, earliest log: {early_log}, current log: {cur_log} failed"
            LOGGER.error(err_msg)
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[lis_bak_cmd, err_msg])
        LOGGER.info(f'Get data backup end time: {end_time}, earliest log: {early_log}, current log: {cur_log} success, '
                    f'start time: {start_time}, database: {db_name}.')
        return end_time, early_log, cur_log

    @staticmethod
    def assemble_regularized_expression(start_time):
        cur_log_begin_reg = r"\s*B\s+D\s+"
        cur_log_end_reg = r"\s+[N|E|O]\s+D\s+(S[\d]{7}\.LOG)\s+(S[\d]{7}\.LOG).*"
        cur_log_info_reg = f"{cur_log_begin_reg}{start_time}\d*{cur_log_end_reg}"
        return cur_log_begin_reg, cur_log_end_reg, cur_log_info_reg

    @staticmethod
    def get_archive_log_path(user_name, dt_name):
        """
        获取数据库归档日志路径
        :param user_name: db2用户
        :param dt_name: 数据库名称
        :return:None 失败。其余成功
        """
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(dt_name)
        query_cmd = f"su - {user_name} -c 'db2 get db cfg for {dt_name}'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        return_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f'Get archive log path failed. return_code: {return_code}, std_out: {std_out}, '
                         f'std_err: {std_err}')
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[query_cmd, std_out])
        split_list = std_out.split('\n')
        result = ""
        for item in split_list:
            if 'First log archive method' not in item:
                continue
            if "DISK" not in item:
                LOGGER.error(f'Get archive log path failed. reasons: Archive log not config, archive config: {item}.')
                raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND,
                                       *[query_cmd, "The configured archive path does not support"])
            result = item.split('=')[1].strip()
            break
        LOGGER.info(f'Get archive log path success. archive path: {result}')
        return result

    @staticmethod
    def get_now_backup_process(user_name, dt_name):
        """
        获取当前备份进度
        :param user_name: db2实例名
        :param dt_name: 数据库名称
        :return: 0, 表示备份还未开始或者备份已经结束
        """
        LOGGER.info(f'Start query backup progress.')
        Db2CommonUtil.check_os_user_with_ex(user_name)
        query_cmd = f"su - {user_name} -c 'db2 list utilities show detail'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        _, std_out, std_err = execute_cmd(cmd=query_cmd, timeout=60, encoding=encoding)
        progress = 0
        # 当前没有备份任务
        if "No data was returned" in std_err or "No data was returned" in std_out:
            LOGGER.info(f'Query backup progress complete with warning. progress: {progress}')
            return progress
        if "没有返回任何数据" in std_err or "没有返回任何数据" in std_out:
            LOGGER.info(f'Query backup progress complete with warning. progress: {progress}')
            return progress
        split_list = std_out.split('\n')
        query_dt_name = ""
        mem_num = 0
        for item in split_list:
            if "Database Name" in item:
                query_dt_name = item.split('=')[1].strip()
                continue
            if "Estimated Percentage Complete" in item:
                if query_dt_name == dt_name:
                    mem_num += 1
                    progress += int(item.split('=')[1].strip())
                continue
        try:
            progress = int(progress / mem_num)
        except Exception as e_info:
            LOGGER.error(f'Query backup progress failed, reason: {e_info}')
            return progress
        LOGGER.info(f'Query backup progress complete. progress: {progress}')
        return progress

    @staticmethod
    def exec_protect_cmd(protect_cmd, db_name, encoding):
        ret_code, std_out, std_err = execute_cmd(protect_cmd, encoding=encoding)
        LOGGER.info(f'exec_protect_cmd return code: {ret_code}, std_out: {std_out}, std_err:{std_err}')
        if "SQL2048N" in std_out and 'Reason code: "6"' in std_out:
            raise ErrCodeException(Db2ErrCode.RUNNING_BACKUP_TASK_EXISTS, db_name)
        # 执行状态错误或者error在输出种则备份失败
        if ret_code != CMDResult.SUCCESS.value or "errors" in std_out:
            LOGGER.error(f'Failed to execute the backup command. return code: {ret_code}, std_out: {std_out}, std_err:'
                         f'{std_err}')
            raise ErrCodeException(Db2ErrCode.DATABASE_CRANKBACK_EXCEPTION, std_out)
        if "备份成功" in std_out:
            return std_out.strip().split("：")[-1]
        else:
            return std_out.strip().split(" ")[-1]

    @staticmethod
    def record_task_result(pid, code, err_msg=None, err_code=None):
        action_result = ActionResult(code=code, message=err_msg)
        if err_code:
            action_result.body_err = err_code
        output_result_file(pid, action_result.dict(by_alias=True))

    @staticmethod
    def safe_get_environ(key):
        if not SysData.SYS_STDIN:
            return ""
        try:
            value_dict = json.loads(SysData.SYS_STDIN)
        except Exception:
            return ""
        value = value_dict.get(key)
        return value if isinstance(value, str) else ""

    @staticmethod
    def get_db_loc_dir(user_name, dt_name):
        Db2CommonUtil.check_os_user_with_ex(user_name)
        query_cmd = f"su - {user_name} -c 'db2 list db directory'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        ret_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if ret_code != CMDResult.SUCCESS.value:
            LOGGER.error(f'Failed to query the database list. ret_code: {ret_code} std_out: {std_out}'
                         f' std_err: {std_err}')
            return False
        split_list = std_out.split('\n')
        find_flag = False
        for item in split_list:
            if "Database name" in item:
                tmp_name = item.split('=')[1].strip()
                if tmp_name == dt_name:
                    find_flag = True
                continue
            elif "数据库名称" in item:
                tmp_name = item.split('=')[1].strip()
                if tmp_name == dt_name:
                    find_flag = True
                continue
            if find_flag and "Local database directory" in item:
                loc_dir = item.split('=')[1].strip()
                LOGGER.info(f"Find the database file directory. database_name: {dt_name} "
                            f"loc dir: {loc_dir}")
                return loc_dir
            elif find_flag and "本地数据库目录" in item:
                loc_dir = item.split('=')[1].strip()
                LOGGER.info(f"Find the database file directory. database_name: {dt_name} "
                            f"loc dir: {loc_dir}")
                return loc_dir
        LOGGER.error(f"Database file directory not found. database_name: {dt_name}")
        return False

    @staticmethod
    def list_db2_nodes(os_user):
        """列出实例的节点ID"""
        node_ids = list()
        Db2CommonUtil.check_os_user_with_ex(os_user)
        query_cmd = f"su - {os_user} -c 'cat ~/sqllib/db2nodes.cfg'"
        encoding = Db2CommonUtil.get_lang_value(os_user)
        ret_cd, stdout, stderr = execute_cmd(query_cmd, encoding=encoding)
        if ret_cd != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute view db2nodes cfg command: {query_cmd} failed, return code: {ret_cd}, "
                         f"out: {stdout}, err: {stderr}.")
            return node_ids
        for tmp_line in stdout.split('\n'):
            mat_node_ret = re.match("\s*(\d+)\s+\S+\s+\d+", tmp_line)
            if mat_node_ret:
                node_ids.append(mat_node_ret.groups()[0])
        LOGGER.info(f"View db2nodes cfg success, node ids: {node_ids}.")
        return node_ids

    @staticmethod
    def get_hostname():
        return socket.gethostname()

    @staticmethod
    def get_partition_no_by_hostname(user_name):
        Db2CommonUtil.check_os_user_with_ex(user_name)
        ht_name = Db2CommonUtil.get_hostname()
        LOGGER.info(f"Start to get the list of partitions on the node.hostname: {ht_name}")
        query_cmd = f"su - {user_name} -c 'cat ~/sqllib/db2nodes.cfg'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        ret_cd, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if ret_cd != CMDResult.SUCCESS.value:
            LOGGER.error(f"Exec cmd failed. cmd: {query_cmd}, return code: {ret_cd} std_out: {std_out} "
                         f"std err: {std_err}")
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[query_cmd, std_out])
        split_list = std_out.split('\n')
        ret_dict = []
        for item in split_list:
            item = item.strip()
            if not item:
                continue
            space_list = item.split()
            if space_list[1] == ht_name:
                ret_dict.append(space_list[0])
        LOGGER.info(f"Success to get the list of partitions on the node.hostname: {ht_name}, "
                    f"part num list: {ret_dict}")
        return ret_dict

    @staticmethod
    def get_partition_no_by_hostname_for_single(user_name):
        Db2CommonUtil.check_os_user_with_ex(user_name)
        ht_name = Db2CommonUtil.get_hostname()
        LOGGER.info(f"Start to get the list of partitions on the node.hostname: {ht_name}")
        query_cmd = f"su - {user_name} -c 'cat ~/sqllib/db2nodes.cfg'"
        ret_cd, std_out, std_err = execute_cmd(query_cmd)
        if ret_cd != CMDResult.SUCCESS.value:
            LOGGER.error(f"Exec cmd failed. cmd: {query_cmd}, return code: {ret_cd} std_out: {std_out} "
                         f"std err: {std_err}")
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[query_cmd, std_out])
        split_list = std_out.split('\n')
        LOGGER.info(f"Content of database node configuration file:{split_list}, len:{len(split_list)}")
        ret_dict = []
        for item in split_list:
            item = item.strip()
            if not item:
                continue
            space_list = item.split()
            if space_list[1] == ht_name:
                ret_dict.append(space_list[0])
        if not ret_dict and len(split_list) == 2:
            item = split_list[0].strip()
            space_list = item.split()
            ret_dict.append(space_list[0])
        LOGGER.info(f"Success to get the list of partitions on the node.hostname: {ht_name}, "
                    f"part num list: {ret_dict}")
        return ret_dict

    @staticmethod
    def get_current_log_chain(os_user, db_name):
        """调用db2pd查询当前日志链ID"""
        log_chain_id = None
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(db_name)
        encoding = Db2CommonUtil.get_lang_value(os_user)
        db2pd_logs_cmd = f'su - {os_user} -c ' \
                         f'"db2 connect to {db_name}; ~/sqllib/adm/db2pd -db {db_name} -logs; db2 terminate"'
        return_code, out, err = execute_cmd(db2pd_logs_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute db2pd logs command with output failed, return code: {return_code}, "
                         f"out: {out}, err: {err}.")
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[db2pd_logs_cmd, out])
        for tmp_line in out.split("\n"):
            tmp_match_chain_id_ret = re.search(r"Log Chain ID\s+(\d+)", tmp_line, flags=re.I)
            if tmp_match_chain_id_ret:
                log_chain_id = tmp_match_chain_id_ret.groups()[0]
                break
        LOGGER.info(f"The current log chain id: {log_chain_id}, database: {db_name}.")
        if log_chain_id is None:
            LOGGER.error(f"The current log chain id is empty")
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND,
                                   *[db2pd_logs_cmd, "The current log chain id is empty"])
        return f'C{str(log_chain_id).rjust(7, "0")}'

    @staticmethod
    def get_need_backup_logs_for_partition(log_param: LogBackupParam,
                                           pre_copy_stop_log, last_archive_log, ptn_num=None,
                                           is_depend_log=True):
        LOGGER.info(f"Getting backup logs for partition {ptn_num}, "
                    f"database: {log_param.db_name}, deploy type: {log_param.deploy_type}, "
                    f"since time: {log_param.since_timestamp}, log chain: {log_param.log_chain}, "
                    f"previous copy stop log: {pre_copy_stop_log}, last archive log: {last_archive_log}, "
                    f"depend log: {is_depend_log}.")
        Db2CommonUtil.check_os_user_with_ex(log_param.user_name)
        Db2CommonUtil.check_injection(str(log_param.since_timestamp), str(ptn_num), log_param.db_name)
        log_scope = f"since {log_param.since_timestamp}" if log_param.since_timestamp else "all"
        if log_param.deploy_type == Db2Const.DPF_CLUSTER_TYPE and ptn_num is not None:
            query_cmd = f"su - {log_param.user_name} -c 'db2_all " \
                        f"\"<<+{ptn_num}<db2 list history archive log {log_scope} for db {log_param.db_name}\"'"
        else:
            query_cmd = f"su - {log_param.user_name} -c 'db2 list history archive log {log_scope} " \
                        f"for db {log_param.db_name}'"
        LOGGER.info(f"Executing list history archive log command: {query_cmd}.")
        encoding = Db2CommonUtil.get_lang_value(log_param.user_name)
        return_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute list history archive log command failed. return_code: {return_code}, "
                         f"out: {std_out}, err: {std_err}.")
            return list()
        return Db2CommonUtil.get_need_bak_log_locs_with_log_chain(
            std_out, log_param.log_chain, pre_copy_stop_log, last_archive_log, is_depend_log=is_depend_log)

    @staticmethod
    def get_need_bak_log_locs_with_log_chain(std_out, log_chain, pre_copy_stop_log, last_archive_log,
                                             is_depend_log=True):
        """
        获取需要备份的日志文件路径列表
        归档日志记录说明：
         Op Obj Timestamp+Sequence Type Dev Earliest Log Current Log  Backup ID
         -- --- ------------------ ---- --- ------------ ------------ --------------
          X  D  20230112200013      1    D  S0000001.LOG C0000000
        Operation：X - Archive log
        Object：D - Database
        Archive log operation types: 1 - Primary log archive method；2 - Secondary log archive method
        Dev：D - DISK
        """
        log_loc_dict = dict()
        arch_log_info_begin_reg = r"\s*X\s+D\s+[\d]{14,}\s+[1|2]\s+D\s+(S[\d]{7}\.LOG)\s+"
        arch_log_info_reg = f"{arch_log_info_begin_reg}{str(log_chain)}\s*"
        tmp_log_name = None
        for tmp_line in std_out.split('\n'):
            tmp_match_log_ret = re.match(arch_log_info_reg, tmp_line)
            if re.match(arch_log_info_reg, tmp_line):
                tmp_log_name = tmp_match_log_ret.groups()[0]
                log_loc_dict[tmp_log_name] = dict()
                continue
            if tmp_log_name and re.search(rf".+Location:.+/{tmp_log_name}$", tmp_line):
                log_loc_dict[tmp_log_name]["location"] = tmp_line.split(':')[-1].strip()
                tmp_log_name = None
            elif tmp_log_name and re.search(rf".+位置：.+/{tmp_log_name}$", tmp_line):
                log_loc_dict[tmp_log_name]["location"] = tmp_line.split('：')[-1].strip()
                tmp_log_name = None
        if is_depend_log:
            return list(v.get("location") for k, v in log_loc_dict.items()
                        if pre_copy_stop_log < k <= last_archive_log and v.get("location"))
        # 日志备份依赖全量副本时需要带上全量副本的日志
        return list(v.get("location") for k, v in log_loc_dict.items()
                    if pre_copy_stop_log <= k <= last_archive_log and v.get("location"))

    @staticmethod
    def get_need_bak_log_locs_without_log_chain(std_out, last_archive_log):
        log_paths = list()
        for item in std_out.split('\n'):
            if "Location:" in item:
                tmp_loc = item.split(':')[-1].strip()
                tmp_log_name = tmp_loc.split("/")[-1]
                if tmp_log_name <= last_archive_log:
                    log_paths.append(tmp_loc)
        return log_paths

    @staticmethod
    def read_json_from_file(file_path):
        if not os.path.isfile(file_path):
            raise Exception(f"File:{file_path} not exist")
        try:
            with open(file_path, "r", encoding=Encoding.INTERNAL_ENCODING) as tmp_file:
                json_dict = json.loads(tmp_file.read())
        except Exception as ex:
            raise Exception("parse param file failed") from ex
        return json_dict

    @staticmethod
    def get_partition_no(user_name, dt_name, cluster_type=None):
        """
        获取DB2数据库分区列表
        :param user_name: DB2实例名
        :param dt_name: DB2 数据库名称
        :param cluster_type: 集群类型
        :return:
        """
        if is_power_rhel_ha(cluster_type) or cluster_type == Db2Const.SINGLE:
            return ['0']
        LOGGER.info(f'Start Obtaining the Location of Latest Archive Logs.')
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(dt_name)
        query_cmd = f"su - {user_name} -c 'db2 connect to {dt_name}; db2 LIST DBPARTITIONNUMS; db2 terminate'"
        return_code, std_out, std_err = execute_cmd(query_cmd)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f'Exec cmd failed. return_code: {return_code}, std_out: {std_out}, '
                         f'std_err: {std_err}')
            return []
        split_list = std_out.split('\n')
        num_list = []
        start_flag = False
        end_flag = False
        for item in split_list:
            if "DATABASE PARTITION NUMBER" in item:
                start_flag = True
                continue
            elif "record(s) selected" in item:
                end_flag = True
            if start_flag and not end_flag:
                num_list.append(item)
        ret_json = []
        for item in num_list:
            item = item.strip()
            if "--" in item or not item:
                continue
            ret_json.append(item.strip())
        return ret_json

    @staticmethod
    def get_archive_log_extend_info(user_name, dt_name, database_type=None):
        """
        获取备份后最新的归档的日志位置，上报副本信息使用
        :param database_type: 数据库类型
        :param user_name: db2 用户名
        :param dt_name: 数据库名称
        :return:
        """
        if database_type == Db2Const.SINGLE:
            num_list = ["0"]
        else:
            num_list = Db2CommonUtil.get_partition_no(user_name, dt_name, database_type)
        ret_json = {
            "extendInfo": {},
            "backupTime": int(time.time()),
            "beginTime": int(time.time()),
            "endTime": int(time.time()),
            "restore_time": "",
            "database_name": dt_name
        }
        for item in num_list:
            tmp_ret = {
                f"{item}": Db2CommonUtil.get_latest_arch_log_local(user_name, dt_name, item, database_type)
            }
            ret_json["extendInfo"].update(tmp_ret)
        return ret_json

    @staticmethod
    def get_latest_arch_log_local(user_name, dt_name, part_num, database_type=None):
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(dt_name, str(part_num))
        if database_type == Db2Const.SINGLE:
            query_cmd = f"su - {user_name} -c 'db2 list history archive log all for db {dt_name}'"
        else:
            query_cmd = f"su - {user_name} -c 'db2_all " \
                        f"\"<<+{part_num}<db2 list history archive log all for db {dt_name}\"'"

        encoding = Db2CommonUtil.get_lang_value(user_name)
        return_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)

        #开始重试查询归档日志操作
        retry_nums = 0
        while retry_nums < 2:
            retry_nums += 1
            if "SQL2155W" not in std_out:
                break
            time.sleep(NumberConst.TEN)
            encoding = Db2CommonUtil.get_lang_value(user_name)
            return_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Exec single cmd failed. return_code: {return_code}, std_out: {std_out} std_err: {std_err}")
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[query_cmd, std_out])
        split_list = std_out.split('\n')
        ret_json = {
            "start_time": 0,
            "end_time": 0,
            "local": ""
        }

        # 选取最后20行,避免不同版本返回格式有误,正常最后一个为13行
        if len(split_list) > 20:
            split_list = split_list[-20:]

        for item in split_list:
            if "Start Time" in item:
                tmp_list = item.split(':')
                ret_json["start_time"] = tmp_list[1].strip()
            if "End Time" in item:
                tmp_list = item.split(':')
                ret_json["end_time"] = tmp_list[1].strip()
            if "Location" in item:
                tmp_list = item.split(':')
                ret_json["local"] = tmp_list[2].strip()
        return ret_json

    @staticmethod
    def create_node_dir_for_dpf_log_backup(os_user, log_path, part_no_list):
        user_id = Db2CommonUtil.get_os_user_id_by_os_user(os_user)
        group_id = Db2CommonUtil.get_group_id_by_os_user(os_user)
        for num in part_no_list:
            tmp_nd_dir = os.path.realpath(os.path.join(log_path, f'NODE{str(num).rjust(4, "0")}'))
            Db2CommonUtil.check_path_list(tmp_nd_dir)
            if not os.path.exists(tmp_nd_dir):
                os.mkdir(tmp_nd_dir)
                os.lchown(tmp_nd_dir, user_id, group_id)

    @staticmethod
    def check_backup_copy_verify(user_name, real_copy_name, check_backup_file):
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(real_copy_name)
        Db2CommonUtil.check_path_list(check_backup_file)
        list_db_cmd = f'su - {user_name} -c "db2ckbkp -h {real_copy_name} > {check_backup_file}"'
        LOGGER.info(f"Executing check backup copy command: {list_db_cmd}.")
        # 存在网络不稳定导致校验报错的情况，增加重试
        for i in range(4):
            return_code, _, std_err = execute_cmd(list_db_cmd)
            if return_code == CMDResult.SUCCESS.value:
                break
            LOGGER.error(f"Execute check db2 backup copy verify fail, retry times: {str(i)}, "
                         f"return code: {return_code}, err: {std_err}.")
            if i == NumberConst.THREE:
                return "false"
            time.sleep(NumberConst.FIVE)
        with open(check_backup_file, "r") as file_read:
            copy_data = file_read.read()
        if "successful" in copy_data:
            if check_backup_file and os.path.isfile(check_backup_file):
                os.remove(check_backup_file)
            return "true"
        if check_backup_file and os.path.isfile(check_backup_file):
            os.remove(check_backup_file)
        return "false"

    @staticmethod
    def check_all_copy_verify(user_name, backup_path, bak_time, cluster_type, file_name):
        is_exists_file = False
        tmp_real_copy_name = []
        for i in os.listdir(backup_path):
            if bak_time not in i:
                # 如果存在副本路径下非副本文件直接返回错误
                LOGGER.error(f"Non-copy files({i}) exist in the copy path.")
                return "false"
            real_copy_name = os.path.join(backup_path, i)
            os.chmod(real_copy_name, 0o755)
            if cluster_type == Db2Const.DPF_CLUSTER_TYPE:
                is_exists_file = True
                check_result = Db2CommonUtil.check_backup_copy_verify(user_name, real_copy_name, file_name)
                if "false" == check_result:
                    LOGGER.error(f"{i} is not can use.")
                    return "false"
            else:
                tmp_real_copy_name.append(real_copy_name)
        # 多路径生成的副本存在多个，传入校验全路径副本 "xxx xxx"
        if is_exists_file:
            return "true"
        check_result = Db2CommonUtil.check_backup_copy_verify(user_name, " ".join(tmp_real_copy_name), file_name)
        if "true" == check_result:
            return "true"
        LOGGER.error(f" {' '.join(tmp_real_copy_name)} is not can use.")
        return "false"

    @staticmethod
    def get_db_names(os_user) -> list:
        """获取所有的数据库名称"""
        Db2CommonUtil.check_os_user_with_ex(os_user)
        db_names = list()
        list_db_cmd = f'su - {os_user} -c "db2 list db directory"'
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, std_out, std_err = execute_cmd(list_db_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute list db directory command failed.")
            raise Exception("Execute list db command failed")
        if "System Database Directory" not in std_out:
            LOGGER.error("The database directory information is incorrect.")
            raise Exception("The database directory information is incorrect")
        for line in std_out.split("\n"):
            if re.match(r"\s*Database name\s*=", str(line)):
                db_names.append(str(line).split("=")[1].strip())
        return db_names

    @staticmethod
    def chown_user_path(file_path, os_user):
        os.chown(file_path,
                 Db2CommonUtil.get_os_user_id_by_os_user(os_user),
                 Db2CommonUtil.get_group_id_by_os_user(os_user))

    @staticmethod
    def chown_user_id_path(file_path, user_id, group_id):
        os.chown(file_path, user_id, group_id)

    @staticmethod
    def clear_db_archive_dir_before_restore(os_user, db_name):
        """恢复前清空数据库的归档目录"""
        log_arch_meth_val = Db2CommonUtil.get_log_arch_meth_val_of_db(os_user, db_name)
        if not log_arch_meth_val:
            return
        full_db_arch_dir = os.path.realpath(os.path.join(log_arch_meth_val, str(os_user), str(db_name)))
        if not os.path.isdir(full_db_arch_dir):
            LOGGER.warning(f"The database archive directory: {full_db_arch_dir} is not directory.")
            return
        clean_dir(full_db_arch_dir)
        LOGGER.info(f"Delete database archive directory: {full_db_arch_dir} success.")

    @staticmethod
    def get_database_partitionnums(user_name, db_name):
        """
        获取数据库的分区号
        :param user_name: 用户名
        :param db_name: 数据库名
        :return: 分区字符串，例: "0,1,2,3"
        """
        LOGGER.info(f"Start to query database partition nums.")
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(db_name)
        query_cmd = f"su - {user_name} -c 'db2 connect to {db_name};db2 list dbpartitionnums;db2 terminate'"
        return_code, std_out, std_err = execute_cmd(query_cmd)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Query database partition nums failed, err info: {std_err}.")
            return ""
        split_list = std_out.split('\n')
        start_flag = False
        ret = ""
        for line_str in split_list:
            tmp_str = line_str.strip()
            if not tmp_str:
                continue
            if "record(s) selected." in tmp_str:
                break
            elif "记录已选择" in tmp_str:
                break
            if "----------------" in tmp_str:
                start_flag = True
                continue
            if start_flag:
                ret = ret + f"{tmp_str},"
        LOGGER.info(f"Query database partition nums complete. Database name: {db_name}, Partition nums: {ret}")
        return ret.strip(',')

    @staticmethod
    def parse_cur_log_info(check_output):
        chain_id, cur_log_id = "", ""
        for line in check_output.split("\n"):
            if chain_id and cur_log_id:
                break
            if not chain_id and re.match(r"\s*logChainId\s*=", str(line)):
                chain_id = str(str(line).split("=")[1]).strip()
                continue
            if not cur_log_id and re.match(r"\s*Current Log Infor\s*=", str(line)):
                cur_log_id = str(str(line).split("=")[1].split(",")[0]).strip()
        log_chain = f'C{str(chain_id).rjust(7, "0")}' if chain_id else ""
        cur_log = f'S{str(cur_log_id).rjust(7, "0")}.LOG' if cur_log_id else ""
        return log_chain, cur_log

    @staticmethod
    def execute_echo_result_command(sh_cmd, timeout=-1, encoding="utf-8"):
        """执行带“echo result=$?”的命令"""
        ssh = pexpect.spawn(sh_cmd, encoding=encoding)
        try:
            index = ssh.expect([r'result=0', pexpect.EOF, pexpect.TIMEOUT], timeout=timeout)
            if index != 0:
                LOGGER.error(f"Execute shell command: {sh_cmd} failed, index: {index}.")
                raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[sh_cmd, ssh.before])
        finally:
            if ssh:
                ssh.close()

    @staticmethod
    def check_file_path_has_existed(file_path_list: list):
        # 检查文件是否存在
        for i in file_path_list:
            if not os.path.exists(i):
                LOGGER.warning(f"The file: {i} does not exists.")
                return False
        return True

    @staticmethod
    def check_process_service(db2_username):
        """检查db2sysc进程是否运行"""
        LOGGER.info(f"Begin to check process service:{db2_username}")
        pid_list = psutil.pids()
        for os_pid in pid_list:
            try:
                pid_name = psutil.Process(os_pid).username()
            except Exception:
                LOGGER.debug(f"This is interim process,pid is {os_pid} ")
                continue
            if pid_name != db2_username:
                continue
            try:
                tmp_cmd = psutil.Process(os_pid).cmdline()
                LOGGER.info(f"tmp_cmd:{tmp_cmd}")
            except Exception:
                LOGGER.debug(f"This is interim process,pid is {os_pid} ")
                continue
            if tmp_cmd and "db2sysc" in tmp_cmd[0]:
                return True
        return False

    @staticmethod
    def check_table_from_tablespace(user_name, database_name, table_space_list: list):
        # 检查表空间所拥有的所有表
        LOGGER.info("Check table from table_space!")
        table_list = []
        table_value_list = []
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(database_name)
        for table in table_space_list:
            Db2CommonUtil.check_injection(table)
            var_cmd = f"su - {user_name} -c \"db2 connect to {database_name};" \
                      f"db2 select tabname from syscat.tables where TBSPACE=\\'{table}\\';db2 list tables\""
            encoding = Db2CommonUtil.get_lang_value(user_name)
            return_code, std_out, std_err = execute_cmd(var_cmd, encoding=encoding)
            if not return_code:
                LOGGER.error("Query table failed!")
                return "", ""
            table_list_for_tablespace, value_table_list = Db2CommonUtil.parse_table_space_list(std_out)
            table_list.extend(table_list_for_tablespace)
            table_value_list.extend(value_table_list)
        return table_list, table_value_list

    @staticmethod
    def parse_table_space_list(std_out: str):
        split_list = std_out.split('\n')
        key_flag = False
        value_flag = False
        key_list = []
        value_list = []
        value_table_list = []
        for line_str in split_list:
            tmp = line_str.strip()
            if not tmp:
                continue
            if "TABNAME" in tmp:
                key_flag = True
                continue
            if "-------" in tmp:
                continue
            if key_flag and "record(s) selected" in tmp:
                key_flag = False
                continue
            if "Schema" in tmp:
                value_flag = True
                continue
            if value_flag and "record(s) selected" in tmp:
                break
            if key_flag:
                key_list.append(tmp)
                continue
            if value_flag:
                space_list = tmp.split()
                if space_list[0] in key_list:
                    value_list.append(space_list[3])
                    value_table_list.append(space_list[0])
        return value_list, value_table_list

    @staticmethod
    def get_database_archive_log_path(user_name, db_name):
        LOGGER.info(f"Start get database {db_name} archive log path.")
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(db_name)
        query_cmd = f"su - {user_name} -c 'db2 get db cfg for {db_name}'"
        ret, std_out, std_err = execute_cmd(query_cmd)
        if ret != CMDResult.SUCCESS.value:
            LOGGER.error(f"Get database {db_name} archive log path failed.out: {std_out},"
                         f"err: {std_err}")
            return ""
        split_list = std_out.split('\n')
        assign_split = None
        for line_list in split_list:
            tmp_line = line_list.strip()
            if "LOGARCHMETH1" in tmp_line:
                assign_split = tmp_line.split('=')
                break
        if not assign_split or "DISK" not in assign_split[1]:
            LOGGER.error(f"Database {db_name} not open archive log.")
            return ""
        ret = assign_split[1].split(':')[1].strip()
        LOGGER.info(f"Get database {db_name} archive log path complete. path: {ret}")
        return ret

    @staticmethod
    def get_roll_status_time(user_name, db_name):
        """查询前滚状态获取最后一个事务时间（针对单机、powerHA、HADR）"""
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(db_name)
        query_cmd = f"su - {user_name} -c 'db2 rollforward database {db_name} query status using local time'"
        ret, std_out, std_err = execute_cmd(query_cmd)
        if ret != CMDResult.SUCCESS.value:
            LOGGER.error(f"Query rollforward status failed. ret: {ret}, out: {std_out}, err: {std_err}")
            raise Exception("Query rollforward status failed.")
        for line in std_out.split('\n'):
            if "Last committed transaction" in line:
                ret = line.split('=')[1].split()[0].strip()
                LOGGER.info(f"Find rollforward time. time: {ret}")
                return ret
        raise Exception("Get rollforward status failed.")

    @staticmethod
    def check_is_primary(user_name, db_name):
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(db_name)
        query_cmd = f"su - {user_name} -c 'db2pd -db {db_name} -hadr'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        ret_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if ret_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Query hadr status cmd exec failed.")
            return False
        split_list = std_out.split('\n')
        primary_flag = False
        connect_flag = False
        for line_str in split_list:
            tmp_str = line_str.strip()
            if "HADR_ROLE" in tmp_str and "PRIMARY" in tmp_str:
                primary_flag = True
                continue
            if "HADR_CONNECT_STATUS" in tmp_str:
                if "CONNECTED" in tmp_str:
                    connect_flag = True
                break
        if primary_flag and connect_flag:
            LOGGER.info("This node is primary node.")
            return True
        LOGGER.info("This node is not primary node.")
        return False

    @staticmethod
    def get_agent_id():
        """
        功能描述：获取Agent ID
        """
        LOGGER.info(f"Start get agent id.")
        agent_file_path = ParamConstant.HOST_SN_FILE_PATH
        if not agent_file_path or not os.path.exists(agent_file_path):
            LOGGER.error("Get agent id failed.")
            return ""
        with open(agent_file_path, "r", encoding=Encoding.INTERNAL_ENCODING) as f_handle:
            text = f_handle.read()
        LOGGER.info(f"Get agent id complete. Agent id: {text}")
        return text.strip("\n")

    @staticmethod
    def check_agent_id_is_same(copy_info_dict):
        LOGGER.info(f"Start check agent id is same")
        now_agent_id = Db2CommonUtil.get_agent_id()
        LOGGER.info(f"Now agent id is : {now_agent_id}")
        pre_agent_id = copy_info_dict.get("extendInfo", {}).get("uuid", "")
        LOGGER.info(f"Pre agent id is : {pre_agent_id}")
        if now_agent_id == pre_agent_id:
            return True
        return False

    @staticmethod
    def domain_conversion_ip(domain_name):
        try:
            host = socket.gethostbyname(domain_name)
        except Exception as ex:
            LOGGER.error("Query host failed!")
            return ""
        else:
            return host

    @staticmethod
    def get_designated_mount_path(rep_type, repositories):
        if not repositories:
            LOGGER.error(f"Repositories is null. rep_type:{rep_type}")
            return ""
        for repository in repositories:
            if repository.get(ParamField.REPOSITORY_TYPE, "") != rep_type:
                continue
            path_array = repository.get(ParamField.PATH, [])
            if len(path_array) <= 0 or not path_array[0]:
                continue
            for path in path_array:
                if common.check_path_legal(path, Db2Const.MOUNT_PATH):
                    return path
        return ""

    @staticmethod
    def verify_path_trustlist(path):
        try:
            real_path = os.path.realpath(path)
        except Exception as error:
            return False
        if common.check_command_injection(real_path):
            return False
        path_trustlist = [
            Db2Const.CACHE_MOUNT_PATH,
            Db2Const.MOUNT_PATH
        ]
        for tmp_path in path_trustlist:
            if real_path.startswith(tmp_path):
                return True
        return False

    @staticmethod
    def get_database_local_path(user_name, db_name):
        Db2CommonUtil.check_os_user_with_ex(user_name)
        query_cmd = f"su - {user_name} -c 'db2 list db directory'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        ret_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)

        if ret_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Query hadr status cmd exec failed.")
            return ""
        split_list = std_out.split('\n')
        find_flag = False
        for lines in split_list:
            line_tmp = lines.strip()
            if not line_tmp:
                continue
            if "Database name" in line_tmp and line_tmp.split('=')[1].strip() == db_name:
                find_flag = True
                continue
            elif "数据库名称" in line_tmp and line_tmp.split('=')[1].strip() == db_name:
                find_flag = True
                continue
            if "Local database directory" in line_tmp and find_flag:
                ret = line_tmp.split('=')[1].strip()
                LOGGER.info(f"Find local database path: {ret}")
                return ret
            elif "本地数据库目录" in line_tmp and find_flag:
                ret = line_tmp.split('=')[1].strip()
                LOGGER.info(f"Find local database path: {ret}")
                return ret
        LOGGER.error(f"Not find local database path: {db_name}")
        return ""

    @staticmethod
    def check_log_continuity(user_name, db_name, timestamp):
        """
        检查日志是否连续
        """
        Db2CommonUtil.check_os_user_with_ex(user_name)
        Db2CommonUtil.check_injection(db_name, str(timestamp))
        query_cmd = f"su - {user_name} -c 'db2 list history archive log since {timestamp} for db {db_name}'"
        encoding = Db2CommonUtil.get_lang_value(user_name)
        ret_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if ret_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Query history log failed.")
            return False
        split_list = std_out.split('\n')
        for lines in split_list:
            line_tmp = lines.strip()
            if "Location" in line_tmp:
                file_name = line_tmp.split(':')[2].strip()
                if not os.path.exists(file_name):
                    LOGGER.error(f"File not exist, {file_name}")
                    return False
        return True

    @staticmethod
    def modify_path_permissions(copy_path):
        ret = common.check_path_legal(copy_path, Db2Const.MOUNT_PATH)
        if not ret:
            raise Exception("The path is not in the white list.")
        os.chmod(copy_path, 0o755)
        for path, child_path, copy_file in os.walk(copy_path):
            for tmp_path in child_path:
                tmp_path_file = os.path.join(path, tmp_path)
                os.chmod(tmp_path_file, 0o755)
            for tmp_copy in copy_file:
                tmp_copy_file = os.path.join(path, tmp_copy)
                os.chmod(tmp_copy_file, 0o644)
        LOGGER.info(f"Succeeded modify copy permissions.")

    @staticmethod
    def mount_path_on_aix(src_path, dest_path):
        """
        在AIX系统上挂载路径
        :param src_path: 源路径
        :param dest_path: 目标路径
        :return: None
        """
        Db2CommonUtil.umount_path_on_aix(dest_path)
        if check_command_injection_exclude_quote(src_path):
            LOGGER.error(f"The path: {src_path} contains special characters.")
            raise ErrCodeException(Db2ErrCode.PARAMS_IS_INVALID, message="The path contains special characters.")
        if not check_path_legal(src_path, Db2Const.MOUNT_PATH):
            LOGGER.error(f"The path: {src_path} is not in the trust list.")
            raise ErrCodeException(Db2ErrCode.PARAMS_IS_INVALID, message="The path is not in the trust list.")
        if not os.path.isdir(dest_path):
            if os.path.isfile(dest_path):
                os.remove(dest_path)
            try:
                os.makedirs(dest_path, 0x755)
            except Exception as ex:
                LOGGER.error(f"Make directory: {dest_path} failed.")
                raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND,
                                       *[f"mkdir -p {dest_path}", "Make directory failed."]) from ex
        else:
            clean_dir(dest_path)
        mount_cmd = f"mount -f {src_path} {dest_path}"
        LOGGER.info(f"Executing mount command: {mount_cmd}.")
        return_code, std_out, std_err = execute_cmd(mount_cmd)
        if return_code != "0":
            LOGGER.error(f"Execute mount command failed, code: {return_code}, out: {std_out}, err: {std_err}.")
            raise ErrCodeException(Db2ErrCode.FAILED_EXECUTE_COMMAND, *[mount_cmd, std_out])
        LOGGER.info("Execute mount command success.")

    @staticmethod
    def umount_path_on_aix(dest_path):
        """
        在AIX系统上解挂载路径
        :param dest_path: 目标路径
        :return: bool True: 成功；False：失败
        """
        if not os.path.exists(dest_path):
            LOGGER.warning(f"The path: {dest_path} does not exist when trying to umount.")
            return True
        umount_bind_cmd = f"umount -f {dest_path}"
        LOGGER.info(f"Executing umount command: {umount_bind_cmd}.")
        return_code, std_out, std_err = execute_cmd(umount_bind_cmd)
        if return_code != "0":
            LOGGER.warning(f"Execute umount path: {dest_path} command failed, code: {return_code}, "
                           f"out: {std_out}, err: {std_err}.")
            return False
        LOGGER.info(f"Execute umount path: {dest_path} command success.")
        return True

    @staticmethod
    def parse_sql_result_to_list(std_out: str):
        split_list = std_out.split('\n')
        LOGGER.info(f"split_list: {split_list}")
        result_list = []
        value_flag = False
        for line_str in split_list:
            tmp_str = line_str.strip()
            if not tmp_str:
                continue
            if "-------" in tmp_str:
                value_flag = True
                continue
            if not value_flag:
                continue
            if "record(s) selected" in tmp_str:
                break
            if "记录已选择" in tmp_str:
                break
            result_list.append(tmp_str.split())
        LOGGER.info(f"result_list: {result_list}")
        return result_list

    @staticmethod
    def get_table_info(os_user, db_name, table_space_list: list):
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(db_name)
        Db2CommonUtil.check_injection(*table_space_list)
        table_spaces_str = '\\\',\\\''.join(table_space_list)
        table_spaces_str = f"\\\'{table_spaces_str}\\\'"
        sql_cmd = f'db2 select a.TABNAME,a.CREATE_TIME from syscat.tables as a, syscat.tablespaces as b ' \
                  f'where b.TBSPACE in \({table_spaces_str}\) and a.TBSPACE = b.TBSPACE'
        var_cmd = f'su - {os_user} -c "db2 connect to {db_name}; {sql_cmd}; db2 terminate;"'
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, std_out, std_err = execute_cmd(var_cmd, encoding=encoding)
        LOGGER.info(f"code: {return_code}, out: {std_out}, err: {std_err}.")
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute list table spaces command failed, code: {return_code}, out: {std_out}, "
                         f"err: {std_err}.")
            raise Exception("Execute list table spaces command failed")
        return Db2CommonUtil.parse_sql_result_to_list(std_out)

    @staticmethod
    def get_tablespace_info(os_user, db_name, table_space_list: list):
        Db2CommonUtil.check_os_user_with_ex(os_user)
        Db2CommonUtil.check_injection(db_name)
        Db2CommonUtil.check_injection(*table_space_list)
        table_spaces_str = '\\\',\\\''.join(table_space_list)
        table_spaces_str = f"\\\'{table_spaces_str}\\\'"
        sql_cmd = f"db2 select TBSPACE, CREATE_TIME from syscat.tablespaces where TBSPACE in \({table_spaces_str}\)"
        var_cmd = f'su - {os_user} -c "db2 connect to {db_name}; {sql_cmd}; db2 terminate;"'
        encoding = Db2CommonUtil.get_lang_value(os_user)
        return_code, std_out, std_err = execute_cmd(var_cmd, encoding=encoding)
        if return_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Execute list table spaces command failed, code: {return_code}, out: {std_out}, "
                         f"err: {std_err}.")
            raise Exception("Execute list table spaces command failed")
        return Db2CommonUtil.parse_sql_result_to_list(std_out)

    @staticmethod
    def check_tablespace_status(os_user, db_name, table_space_list: list, copy_tablespace_info: set):
        tablespace_info_list = Db2CommonUtil.get_tablespace_info(os_user, db_name, table_space_list)
        tmp_dict = {k: v for [k, v] in tablespace_info_list}
        tablespace_info = set(zip(tmp_dict.keys(), tmp_dict.values()))
        tablespace_match_fail_set = set()
        if copy_tablespace_info != tablespace_info:
            tablespace_match_fail_set = copy_tablespace_info.difference(tablespace_info)
        return tablespace_match_fail_set

    @staticmethod
    def check_table_status(os_user, db_name, table_space_list: list, copy_table_info: set):
        table_info_list = Db2CommonUtil.get_table_info(os_user, db_name, table_space_list)
        tmp_dict = {k: v for [k, v] in table_info_list}
        table_info = set(zip(tmp_dict.keys(), tmp_dict.values()))
        table_match_fail_set = set()
        if copy_table_info != table_info:
            table_match_fail_set = copy_table_info.difference(table_info)
        return table_match_fail_set

    @staticmethod
    def proactively_report_progress(pid, job_detail):
        """
        功能描述：主动上报进度
        参数：无
        @pid: 任务唯一标识
        @job_detail: 任务详情
        """
        input_file = os.path.join(ParamConstant.RESULT_PATH, str(uuid.uuid4()) + '_input')
        output_file = os.path.join(ParamConstant.RESULT_PATH, str(uuid.uuid4()) + '_output')
        json_str = json.dumps(job_detail.dict(by_alias=True))
        LOGGER.debug(f"Report job detail: {json_str}, input_file: {input_file}.")
        flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
        modes = stat.S_IWUSR | stat.S_IRUSR | stat.S_IXUSR
        with os.fdopen(os.open(input_file, flags, modes), 'w') as jsonfile:
            jsonfile.write(json_str)
        try:
            Db2CommonUtil.exec_rpc_tool_cmd(RpcToolApiName.REPORT_JOB_DETAILS, input_file, output_file)
        except Exception as ex:
            with open(output_file, 'r', encoding=Encoding.INTERNAL_ENCODING) as file_read:
                LOGGER.error(f'{pid} report result: {file_read.readlines()}, error: {ex}.')
        finally:
            os.remove(input_file)
            os.remove(output_file)

    @staticmethod
    def check_injection(*args):
        expression = "[|;&$><`'\"!+\n]"
        for arg in args:
            if arg and re.search(expression, arg):
                raise Exception("Invalid argument")

    @staticmethod
    def check_path_list(*paths):
        for path in paths:
            if path and not Db2CommonUtil.verify_path_trustlist(path):
                LOGGER.error(f"The path: {path} is invalid.")
                raise Exception("Invalid path")

    @staticmethod
    def deactivate_db_status(os_user, db_name):
        """
        功能描述：去激活数据库状态
        参数：无
        @os_user: 用户名
        @db_name: 数据库名称
        """
        if common.check_command_injection_ex_quote_space(db_name):
            raise Exception("Param of node(db_name) invalid")
        query_cmd = f"su - {os_user} -c 'db2 deactivate db {db_name} '"
        LOGGER.info(f"Start to deactivate database: {query_cmd}.")
        encoding = Db2CommonUtil.get_lang_value(os_user)
        ret, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        LOGGER.info(f"Deactivate database completed: {ret}, out info: {std_out}, err info: {std_err}.")
        if ret != CMDResult.SUCCESS.value:
            LOGGER.error(f"Deactivate db failed.")

    @staticmethod
    def activate_db_status(os_user, db_name):
        """
        功能描述：激活数据库状态
        参数：无
        @os_user: 用户名
        @db_name: 数据库名称
        """
        if common.check_command_injection_ex_quote_space(db_name):
            raise Exception("Param of node(db_name) invalid")
        query_cmd = f"su - {os_user} -c 'db2 activate db {db_name} '"
        encoding = Db2CommonUtil.get_lang_value(os_user)
        LOGGER.info(f"Start to activate database: {query_cmd}.")
        ret, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        LOGGER.info(f"Activate database completed: {ret}, out info: {std_out}, err info: {std_err}.")
        if ret != CMDResult.SUCCESS.value:
            LOGGER.error(f"Activate db failed.")

    @staticmethod
    def get_database_size(os_user, db_name):
        """
        功能描述：激活数据库状态
        参数：无
        @os_user: 用户名
        @db_name: 数据库名称
        """
        if common.check_command_injection_ex_quote_space(db_name):
            raise Exception("Param of node(db_name) invalid")
        query_cmd = (f"su - {os_user} -c 'db2 connect to {db_name}; "
                     f"db2 \"call get_dbsize_info(?,?,?,-1)\"; db2 terminate'")
        encoding = Db2CommonUtil.get_lang_value(os_user)
        ret_code, std_out, std_err = execute_cmd(query_cmd, encoding=encoding)
        if ret_code != CMDResult.SUCCESS.value:
            LOGGER.error(f"Query database {db_name} size cmd exec failed.")
            return 0
        split_list = std_out.split('\n')
        find_flag = False
        for lines in split_list:
            line_tmp = lines.strip()
            if not line_tmp:
                continue
            if "DATABASESIZE" in line_tmp:
                find_flag = True
                continue
            if "Parameter Value" in line_tmp and find_flag:
                ret = line_tmp.split(':')[1].strip()
                LOGGER.info(f"Find database size: {ret}")
                return ret
        LOGGER.error(f"query database {db_name} size fail")
        return 0

    @staticmethod
    def check_os_user_with_ex(os_user):
        try:
            pwd.getpwnam(os_user)
        except TypeError as ex:
            LOGGER.error(f"Invalid user: {os_user}, err: {ex}.")
            raise ErrCodeException(Db2ErrCode.OS_USER_NOT_EXISTS) from ex
        except KeyError as ex:
            LOGGER.error(f"Invalid user: {os_user}, err: {ex}.")
            raise ErrCodeException(Db2ErrCode.OS_USER_NOT_EXISTS) from ex
        return True


    @staticmethod
    def get_lang_value(db2_user_name):
        Db2CommonUtil.check_os_user_with_ex(db2_user_name)
        ex_cmd = f"su - {db2_user_name} -c 'echo $LANG'"
        return_code, out_info, err_info = execute_cmd(ex_cmd)
        ret = (return_code == "0")
        lang_value = out_info if ret else err_info
        encoding = "utf-8"
        if not ret or not lang_value:
            LOGGER.error(f"Get user encoding fail, ret: {ret}, lang_value: {lang_value} use default encoding utf-8.")
            return encoding
        lang_value = lang_value.strip('\n').strip()
        LOGGER.info(f"lang_value:{lang_value}")
        if check_command_injection(lang_value):
            LOGGER.error(f"Check locale fail, out: {lang_value} use default encoding utf-8.")
            return encoding
        user_locale = lang_value.split(".")
        if len(user_locale) != 2:
            LOGGER.error("The locale format is incorrect use default encoding utf-8.")
            return encoding
        return user_locale[-1]

    @staticmethod
    def execute_cmd_list(cmd_list, encoding='utf-8', timeout=None):
        """
        执行命令列表，前一条命令的输出是后一条命令的输入
        :param cmd_list: 命令列表
        :param encoding: 指定的编码方式
        :param timeout: 执行命令超时
        :return: (字符串格式返回码, 标准输出, 标准错误)
        """
        ret_code = 1
        std_out = None
        std_err = None
        for tmp_cmd in cmd_list:
            process = subprocess.Popen(shlex.split(tmp_cmd), stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                       stderr=subprocess.PIPE,
                                       encoding=encoding, errors='ignore')
            try:
                std_out, std_err = process.communicate(input=std_out, timeout=timeout)
            except TimeoutError as err:
                process.kill()
                ret_code = 1
                std_out = ""
                std_err = str(err)
                break
            except Exception as err:
                LOGGER.error(f"err:{err}")
            else:
                ret_code = process.returncode
        return str(ret_code), std_out, std_err

    @staticmethod
    def execute_cmd_oversize_return_value(cmd, encoding="utf-8"):
        """
        执行shell命令，不支持管道和重定向, 针对返回值超大的情况使用此命令
        :param cmd: shell命令
        :return:执行返回码，正确信息，错误信息
        """
        exitcode = 0
        try:
            data = subprocess.check_output(shlex.split(cmd), encoding=encoding, shell=False, text=True,
                                           stderr=subprocess.STDOUT)
        except subprocess.CalledProcessError as ex:
            data = ex.output
            exitcode = ex.returncode
        except Exception:
            # 存在utf-8 解析不了的返回信息。
            data = "cmd error"
            exitcode = 127
        if data[-1:] == '\n':
            data = data[:-1]
        return str(exitcode), data, data
